package com.ly.hb.p1.kafka;

import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.ly.hb.p1.entity.WisdomTransferLine;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.h2.util.json.JSONObject;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

@Component
public class ConsumerListener {
    private static AtomicInteger totalCount = new AtomicInteger(0);
//    @KafkaListener(id = "consumer1",topics = "${kafka.topic.test}", groupId = "testGroup1")
//    public void onMessage(String message){
//        System.out.println("11-testGroup1:" + message);
//    }
//
//    @KafkaListener(id = "consumer2",topics = "${kafka.topic.test}", groupId = "testGroup1")
//    public void onMessage2(String message){
//        System.out.println("12-testGroup1:" + message);
//    }

    @KafkaListener(id = "consumer3", groupId = "testGroup2",
            topicPartitions = {
                @TopicPartition(topic = "${kafka.topic.test}",
                                // partitions = "0",
                                partitionOffsets = {
                                    @PartitionOffset(partition = "0", initialOffset = "0"),
                                    @PartitionOffset(partition = "1", initialOffset = "0"),
                                    @PartitionOffset(partition = "2", initialOffset = "0")
                                }
                )
            }
    )
    public void onMessage21(List<ConsumerRecord<?, ?>> records) {
        int totalCount = ConsumerListener.totalCount.addAndGet(records.size());
        System.out.println("21-testGroup2: 批量 batchSize:"+ records.size() +", totalCount:" + totalCount);
        for (ConsumerRecord record : records) {
            System.out.println("21-testGroup2: topic:"+record.topic()+"|partition:"+record.partition()+"|offset:"+record.offset()+"|value:"+record.value());
            String value = String.valueOf(record.value());
            Gson gson = new Gson();
            WisdomTransferLine wisdomTransferLine = gson.fromJson(value, WisdomTransferLine.class);
        }
    }
}
